/*

 * Licensed to the Apache Software Foundation (ASF) under one

 * or more contributor license agreements.  See the NOTICE file

 * distributed with this work for additional information

 * regarding copyright ownership.  The ASF licenses this file

 * to you under the Apache License, Version 2.0 (the

 * "License"); you may not use this file except in compliance

 * with the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

package com.bff.gaia.unified.runners.core.triggers;



import com.bff.gaia.unified.runners.core.MergingStateAccessor;

import com.bff.gaia.unified.runners.core.StateAccessor;

import com.bff.gaia.unified.sdk.state.TimeDomain;

import com.bff.gaia.unified.sdk.transforms.GroupByKey;

import com.bff.gaia.unified.sdk.transforms.windowing.BoundedWindow;

import com.bff.gaia.unified.sdk.transforms.windowing.Window;

import com.bff.gaia.unified.sdk.transforms.windowing.WindowFn;

import com.bff.gaia.unified.vendor.guava.com.google.common.base.Joiner;

import org.joda.time.Instant;



import javax.annotation.Nullable;

import java.io.Serializable;

import java.util.List;

import java.util.Objects;



/**

 * {@code Trigger}s control when the elements for a specific key and window are output. As elements

 * arrive, they are put into one or more windows by a {@link Window} transform and its associated

 * {@link WindowFn}, and then passed to the associated {@code Trigger} to determine if the {@code

 * Window}s contents should be output.

 *

 * <p>See {@link GroupByKey} and {@link Window} for more information

 * about how grouping with windows works.

 *

 * <p>The elements that are assigned to a window since the last time it was fired (or since the

 * window was created) are placed into the current window pane. Triggers are evaluated against the

 * elements as they are added. When the root trigger fires, the elements in the current pane will be

 * output. When the root trigger finishes (indicating it will never fire again), the window is

 * closed and any new elements assigned to that window are discarded.

 *

 * <p>Several predefined {@code Trigger}s are provided:

 *

 * <ul>

 *   <li>{@link AfterWatermarkStateMachine} for firing when the watermark passes a timestamp

 *       determined from either the end of the window or the arrival of the first element in a pane.

 *   <li>{@link AfterProcessingTimeStateMachine} for firing after some amount of processing time has

 *       elapsed (typically since the first element in a pane).

 *   <li>{@link AfterPaneStateMachine} for firing off a property of the elements in the current

 *       pane, such as the number of elements that have been assigned to the current pane.

 * </ul>

 *

 * <p>In addition, {@code Trigger}s can be combined in a variety of ways:

 *

 * <ul>

 *   <li>{@link RepeatedlyStateMachine#forever} to create a trigger that executes forever. Any time

 *       its argument finishes it gets reset and starts over. Can be combined with {@link

 *       TriggerStateMachine#orFinally} to specify a condition that causes the repetition to stop.

 *   <li>{@link AfterEachStateMachine#inOrder} to execute each trigger in sequence, firing each (and

 *       every) time that a trigger fires, and advancing to the next trigger in the sequence when it

 *       finishes.

 *   <li>{@link AfterFirstStateMachine#of} to create a trigger that fires after at least one of its

 *       arguments fires. An {@link AfterFirstStateMachine} trigger finishes after it fires once.

 *   <li>{@link AfterAllStateMachine#of} to create a trigger that fires after all least one of its

 *       arguments have fired at least once. An {@link AfterAllStateMachine} trigger finishes after

 *       it fires once.

 * </ul>

 *

 * <p>Each trigger tree is instantiated per-key and per-window. Every trigger in the tree is in one

 * of the following states:

 *

 * <ul>

 *   <li>Never Existed - before the trigger has started executing, there is no state associated with

 *       it anywhere in the system. A trigger moves to the executing state as soon as it processes

 *       in the current pane.

 *   <li>Executing - while the trigger is receiving items and may fire. While it is in this state,

 *       it may persist book-keeping information to persisted state, set timers, etc.

 *   <li>Finished - after a trigger finishes, all of its book-keeping data is cleaned up, and the

 *       system remembers only that it is finished. Entering this state causes us to discard any

 *       elements in the buffer for that window, as well.

 * </ul>

 *

 * <p>Once finished, a trigger cannot return itself back to an earlier state, however a composite

 * trigger could reset its sub-triggers.

 *

 * <p>Triggers should not build up any state internally since they may be recreated between

 * invocations of the callbacks. All important values should be persisted using state before the

 * callback returns.

 */

public abstract class TriggerStateMachine implements Serializable {



  /**

   * Interface for accessing information about the trigger being executed and other triggers in the

   * same tree.

   */

  public interface TriggerInfo {



    /**

     * Returns true if the windowing strategy of the current {@code PCollection} is a merging {@link

     * WindowFn}. If true, the trigger execution needs to keep enough information to support the

     * possibility of {@link TriggerStateMachine#onMerge} being called. If false, {@link

     * TriggerStateMachine#onMerge} will never be called.

     */

    boolean isMerging();



    /** Access the executable versions of the sub-triggers of the current trigger. */

    Iterable<ExecutableTriggerStateMachine> subTriggers();



    /** Access the executable version of the specified sub-trigger. */

    ExecutableTriggerStateMachine subTrigger(int subtriggerIndex);



    /** Returns true if the current trigger is marked finished. */

    boolean isFinished();



    /** Return true if the given subtrigger is marked finished. */

    boolean isFinished(int subtriggerIndex);



    /** Returns true if all the sub-triggers of the current trigger are marked finished. */

    boolean areAllSubtriggersFinished();



    /** Returns an iterable over the unfinished sub-triggers of the current trigger. */

    Iterable<ExecutableTriggerStateMachine> unfinishedSubTriggers();



    /** Returns the first unfinished sub-trigger. */

    ExecutableTriggerStateMachine firstUnfinishedSubTrigger();



    /**

     * Clears all keyed state for triggers in the current sub-tree and unsets all the associated

     * finished bits.

     */

    void resetTree() throws Exception;



    /** Sets the finished bit for the current trigger. */

    void setFinished(boolean finished);



    /** Sets the finished bit for the given sub-trigger. */

    void setFinished(boolean finished, int subTriggerIndex);

  }



  /**

   * Interact with properties of the trigger being executed, with extensions to deal with the

   * merging windows.

   */

  public interface MergingTriggerInfo extends TriggerInfo {



    /** Return true if the trigger is finished in any window being merged. */

    boolean finishedInAnyMergingWindow();



    /** Return true if the trigger is finished in all windows being merged. */

    boolean finishedInAllMergingWindows();

  }



  /**

   * Information accessible to all operational hooks in this {@code Trigger}.

   *

   * <p>Used directly in {@link TriggerStateMachine#shouldFire} and {@link

   * TriggerStateMachine#clear}, and extended with additional information in other methods.

   */

  public abstract static class TriggerContext {



    /** Returns the interface for accessing trigger info. */

    public abstract TriggerInfo trigger();



    /** Returns the interface for accessing persistent state. */

    public abstract StateAccessor<?> state();



    /** The window that the current context is executing in. */

    public abstract BoundedWindow window();



    /** Create a sub-context for the given sub-trigger. */

    public abstract TriggerContext forTrigger(ExecutableTriggerStateMachine trigger);



    /**

     * Removes the timer set in this trigger context for the given {@link Instant} and {@link

     * TimeDomain}.

     */

    public abstract void deleteTimer(Instant timestamp, TimeDomain domain);



    /** The current processing time. */

    public abstract Instant currentProcessingTime();



    /** The current synchronized upstream processing time or {@code null} if unknown. */

    @Nullable

    public abstract Instant currentSynchronizedProcessingTime();



    /** The current event time for the input or {@code null} if unknown. */

    @Nullable

    public abstract Instant currentEventTime();

  }



  /**

   * Extended {@link TriggerContext} containing information accessible to the {@link #onElement}

   * operational hook.

   */

  public abstract static class OnElementContext extends TriggerContext {

    /** The event timestamp of the element currently being processed. */

    public abstract Instant eventTimestamp();



    /**

     * Sets a timer to fire when the watermark or processing time is beyond the given timestamp.

     * Timers are not guaranteed to fire immediately, but will be delivered at some time afterwards.

     *

     * <p>As with {@link #state}, timers are implicitly scoped to the current window. All timer

     * firings for a window will be received, but the implementation should choose to ignore those

     * that are not applicable.

     *

     * @param timestamp the time at which the trigger should be re-evaluated

     * @param domain the domain that the {@code timestamp} applies to

     */

    public abstract void setTimer(Instant timestamp, TimeDomain domain);



    /** Create an {@code OnElementContext} for executing the given trigger. */

    @Override

    public abstract OnElementContext forTrigger(ExecutableTriggerStateMachine trigger);

  }



  /**

   * Extended {@link TriggerContext} containing information accessible to the {@link #onMerge}

   * operational hook.

   */

  public abstract static class OnMergeContext extends TriggerContext {

    /**

     * Sets a timer to fire when the watermark or processing time is beyond the given timestamp.

     * Timers are not guaranteed to fire immediately, but will be delivered at some time afterwards.

     *

     * <p>As with {@link #state}, timers are implicitly scoped to the current window. All timer

     * firings for a window will be received, but the implementation should choose to ignore those

     * that are not applicable.

     *

     * @param timestamp the time at which the trigger should be re-evaluated

     * @param domain the domain that the {@code timestamp} applies to

     */

    public abstract void setTimer(Instant timestamp, TimeDomain domain);



    /** Create an {@code OnMergeContext} for executing the given trigger. */

    @Override

    public abstract OnMergeContext forTrigger(ExecutableTriggerStateMachine trigger);



    @Override

    public abstract MergingStateAccessor<?, ?> state();



    @Override

    public abstract MergingTriggerInfo trigger();

  }



  @Nullable

  protected final List<TriggerStateMachine> subTriggers;



  protected TriggerStateMachine(@Nullable List<TriggerStateMachine> subTriggers) {

    this.subTriggers = subTriggers;

  }



  /** Called every time an element is incorporated into a window. */

  public abstract void onElement(OnElementContext c) throws Exception;



  /**

   * Called immediately after windows have been merged.

   *

   * <p>Leaf triggers should update their state by inspecting their status and any state in the

   * merging windows. Composite triggers should update their state by calling {@link

   * ExecutableTriggerStateMachine#invokeOnMerge} on their sub-triggers, and applying appropriate

   * logic.

   *

   * <p>A trigger such as {@link AfterWatermarkStateMachine#pastEndOfWindow} may no longer be

   * finished; it is the responsibility of the trigger itself to record this fact. It is forbidden

   * for a trigger to become finished due to {@link #onMerge}, as it has not yet fired the pending

   * elements that led to it being ready to fire.

   *

   * <p>The implementation does not need to clear out any state associated with the old windows.

   */

  public abstract void onMerge(OnMergeContext c) throws Exception;



  /**

   * Returns {@code true} if the current state of the trigger indicates that its condition is

   * satisfied and it is ready to fire.

   */

  public abstract boolean shouldFire(TriggerContext context) throws Exception;



  /**

   * Adjusts the state of the trigger to be ready for the next pane. For example, a {@link

   * RepeatedlyStateMachine} trigger will reset its inner trigger, since it has fired.

   *

   * <p>If the trigger is finished, it is the responsibility of the trigger itself to record that

   * fact via the {@code context}.

   */

  public abstract void onFire(TriggerContext context) throws Exception;



  /**

   * Called to allow the trigger to prefetch any state it will likely need to read from during an

   * {@link #onElement} call.

   */

  public void prefetchOnElement(StateAccessor<?> state) {

    if (subTriggers != null) {

      for (TriggerStateMachine trigger : subTriggers) {

        trigger.prefetchOnElement(state);

      }

    }

  }



  /**

   * Called to allow the trigger to prefetch any state it will likely need to read from during an

   * {@link #onMerge} call.

   */

  public void prefetchOnMerge(MergingStateAccessor<?, ?> state) {

    if (subTriggers != null) {

      for (TriggerStateMachine trigger : subTriggers) {

        trigger.prefetchOnMerge(state);

      }

    }

  }



  /**

   * Called to allow the trigger to prefetch any state it will likely need to read from during an

   * {@link #shouldFire} call.

   */

  public void prefetchShouldFire(StateAccessor<?> state) {

    if (subTriggers != null) {

      for (TriggerStateMachine trigger : subTriggers) {

        trigger.prefetchShouldFire(state);

      }

    }

  }



  /**

   * Called to allow the trigger to prefetch any state it will likely need to read from during an

   * {@link #onFire} call.

   */

  public void prefetchOnFire(StateAccessor<?> state) {

    if (subTriggers != null) {

      for (TriggerStateMachine trigger : subTriggers) {

        trigger.prefetchOnFire(state);

      }

    }

  }



  /**

   * Clear any state associated with this trigger in the given window.

   *

   * <p>This is called after a trigger has indicated it will never fire again. The trigger system

   * keeps enough information to know that the trigger is finished, so this trigger should clear all

   * of its state.

   */

  public void clear(TriggerContext c) throws Exception {

    if (subTriggers != null) {

      for (ExecutableTriggerStateMachine trigger : c.trigger().subTriggers()) {

        trigger.invokeClear(c);

      }

    }

  }



  public Iterable<TriggerStateMachine> subTriggers() {

    return subTriggers;

  }



  /** Returns whether this performs the same triggering as the given {@code Trigger}. */

  public boolean isCompatible(TriggerStateMachine other) {

    if (!getClass().equals(other.getClass())) {

      return false;

    }



    if (subTriggers == null) {

      return other.subTriggers == null;

    } else if (other.subTriggers == null) {

      return false;

    } else if (subTriggers.size() != other.subTriggers.size()) {

      return false;

    }



    for (int i = 0; i < subTriggers.size(); i++) {

      if (!subTriggers.get(i).isCompatible(other.subTriggers.get(i))) {

        return false;

      }

    }



    return true;

  }



  @Override

  public String toString() {

    String simpleName = getClass().getSimpleName();

    if (getClass().getEnclosingClass() != null) {

      simpleName = getClass().getEnclosingClass().getSimpleName() + "." + simpleName;

    }

    if (subTriggers == null || subTriggers.isEmpty()) {

      return simpleName;

    } else {

      return simpleName + "(" + Joiner.on(", ").join(subTriggers) + ")";

    }

  }



  @Override

  public boolean equals(Object obj) {

    if (this == obj) {

      return true;

    }

    if (!(obj instanceof TriggerStateMachine)) {

      return false;

    }

    TriggerStateMachine that = (TriggerStateMachine) obj;

    return Objects.equals(getClass(), that.getClass())

        && Objects.equals(subTriggers, that.subTriggers);

  }



  @Override

  public int hashCode() {

    return Objects.hash(getClass(), subTriggers);

  }



  /**

   * Specify an ending condition for this trigger. If the {@code until} fires then the combination

   * fires.

   *

   * <p>The expression {@code t1.orFinally(t2)} fires every time {@code t1} fires, and finishes as

   * soon as either {@code t1} finishes or {@code t2} fires, in which case it fires one last time

   * for {@code t2}. Both {@code t1} and {@code t2} are executed in parallel. This means that {@code

   * t1} may have fired since {@code t2} started, so not all of the elements that {@code t2} has

   * seen are necessarily in the current pane.

   *

   * <p>For example the final firing of the following trigger may only have 1 element:

   *

   * <pre>{@code

   * Repeatedly.forever(AfterPane.elementCountAtLeast(2))

   *     .orFinally(AfterPane.elementCountAtLeast(5))

   * }</pre>

   */

  public TriggerStateMachine orFinally(TriggerStateMachine until) {

    return new OrFinallyStateMachine(this, until);

  }

}